package com.ljm.boot.kafka.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @Description
 * @Author Dominick Li
 * @CreateTime 2024/1/16 14:41
 **/
@Slf4j
@RestController
public class KafkaAdminController {

    @Resource
    private AdminClient adminClient;

    /**
     * 创建主题
     */
    @GetMapping("/create/{topicName}")
    public String createTopic(@PathVariable String topicName) throws Exception {

        //需要判主题是否已存在,已存在再创建会报错
        if (getTopicSet().contains(topicName)) {
            return "topicExists ";
        }
        // 创建主题  参数：主题名称、分区数、副本数
        CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(new NewTopic(topicName, 1, (short) 1)));
        result.all().get();
        return "success";
    }

    /**
     * 查看所有主题
     */
    @GetMapping("/listTopic")
    public String listTopic() throws Exception {
        Set<String> set = getTopicSet();
        return String.format("topics[%s]", getTopicSet().stream().collect(Collectors.joining(",")));
    }

    public Set<String> getTopicSet() throws Exception {
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        KafkaFuture<Set<String>> future = listTopicsResult.names();
        return future.get();
    }

}

